package com.yeziji.common.rabbitMq;

import com.rabbitmq.client.Channel;
import com.yeziji.common.business.dataMessage.entity.DataMessageRecordEntity;
import com.yeziji.common.business.dataMessage.service.DataMessageRecordService;
import com.yeziji.constant.DataSyncStatusEnum;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.ReturnedMessage;
import org.springframework.amqp.rabbit.connection.CorrelationData;

import javax.annotation.Resource;
import java.io.IOException;

/**
 * rabbit 主题处理器
 *
 * @author hwy
 * @since 2024/10/26 22:46
 **/
@Slf4j
public abstract class RabbitMqTopicHandler {
    @Resource
    private DataMessageRecordService dataMessageRecordService;

    private boolean dead = false;

    /**
     * 统一处理模板
     *
     * @param message rabbit mq 消息体
     * @param channel 消息 channel
     * @param body    消息
     */
    public void handler(Message message, Channel channel, String body) {
        long deliveryTag = message.getMessageProperties().getDeliveryTag();
        try {
            if (dead) {
                this.deadProcess(message, body);
            } else {
                this.process(message, body);
            }
            channel.basicAck(deliveryTag, false);
        } catch (Exception e) {
            log.error("确认消息异常：{}", e.getMessage());
            this.errorHandler(message, channel);
        }
    }

    /**
     * 统一消息发送处理模板
     *
     * @param ack             消息是否发送成功
     * @param correlationData 关联数据
     * @param message         消息
     */
    public void confirm(boolean ack, CorrelationData correlationData, String message) {
        this.confirmMessage(ack, correlationData);
        if (ack) {
            this.confirmAck(correlationData, message);
        } else {
            this.confirmUnAck(correlationData, message);
        }
    }

    /**
     * 统一消息发送异常处理模板
     *
     * @param returnedMessage 返回的消息
     */
    public void returned(ReturnedMessage returnedMessage) {
        Message message = returnedMessage.getMessage();
        if (message != null) {
            String correlationId = message.getMessageProperties().getCorrelationId();
            if (correlationId != null) {
                this.returnedMessage(new CorrelationData(correlationId));
            }
        }
        this.returnedProcess(returnedMessage);
    }

    /**
     * 默认的消息异常处理
     */
    protected void errorHandler(Message message, Channel channel) {
        try {
            channel.basicReject(message.getMessageProperties().getDeliveryTag(), false);
        } catch (IOException ex) {
            log.error("拒绝消息异常：{}", ex.getMessage());
        }
    }

    /**
     * 確認消息
     *
     * @param ack             rabbitmq 是否成功发送到了交换机
     * @param correlationData 关联数据
     */
    protected void confirmMessage(boolean ack, CorrelationData correlationData) {
        if (correlationData != null) {
            String id = correlationData.getId();
            DataMessageRecordEntity record = dataMessageRecordService.getByFlagId(id);
            if (record != null) {
                record.setStatus(ack ? DataSyncStatusEnum.SUCCESS.getCode() : DataSyncStatusEnum.FAILED.getCode());
                dataMessageRecordService.updateById(record);
            }
        }
    }

    /**
     * 確認消息
     *
     * @param correlationData 关联数据
     */
    protected void returnedMessage(CorrelationData correlationData) {
        this.confirmMessage(false, correlationData);
    }

    /**
     * 获取模板主题
     *
     * @return {@link RabbitMqTopicEnum}
     */
    public abstract RabbitMqTopicEnum getTopicEnum();

    /**
     * 子类自定义实现处理方法
     *
     * @param message 处理消息
     * @param body    处理 body -> 处理反序列化
     * @see com.yeziji.config.rabbitMq.RabbitMqListener#onMessage(Message, Channel)
     */
    public abstract void process(Message message, String body);

    /**
     * 子类自定义处理死信消息
     *
     * @param message 处理消息
     * @param body    处理的 body
     */
    public abstract void deadProcess(Message message, String body);

    /**
     * 消息确认行为
     *
     * @param correlationData 确认关联数据
     * @param message         消息
     */
    public abstract void confirmAck(CorrelationData correlationData, String message);

    /**
     * 消息未确认行为
     *
     * @param correlationData 未确认关联数据
     * @param message         消息
     */
    public abstract void confirmUnAck(CorrelationData correlationData, String message);

    /**
     * 消息异常行为
     *
     * @param returnedMessage 异常返回来的通知消息
     */
    public abstract void returnedProcess(ReturnedMessage returnedMessage);

    public void setDead(boolean dead) {
        this.dead = dead;
    }
}
